package com.group2.edu.realtime.dwd.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.TableLearningPlay;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwdLearningPlay extends BaseApp {

    public static void main(String[] args) {
        new DwdLearningPlay().start(10014,4,"dwd_learning_play", Constant.TOPIC_DWD_TRAFFIC_PLAY);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {

        SingleOutputStreamOperator<TableLearningPlay> maplp = ds.map(
                new MapFunction<String, TableLearningPlay>() {
                    @Override
                    public TableLearningPlay map(String js) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(js);
                        JSONObject common = jsonObject.getJSONObject("common");
                        JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                        Long ts = jsonObject.getLong("ts");
                        String sc = common.getString("sc");
                        String ar = common.getString("ar");
                        String uid = common.getString("uid");
                        String os = common.getString("os");
                        String ch = common.getString("ch");
                        String isNew = common.getString("is_new");
                        String md = common.getString("md");
                        String mid = common.getString("mid");
                        String vc = common.getString("vc");
                        String ba = common.getString("ba");
                        String sid = common.getString("sid");
                        Integer playSec = appVideo.getInteger("play_sec");
                        Integer positionSec = appVideo.getInteger("position_sec");
                        String videoId = appVideo.getString("video_id");
                        return TableLearningPlay.builder()
                                .sourceId(sc)
                                .provinceId(ar)
                                .userId(uid)
                                .operatingSystem(os)
                                .channel(ch)
                                .isNew(isNew)
                                .model(md)
                                .machineId(mid)
                                .versionCode(vc)
                                .brand(ba)
                                .sessionId(sid)
                                .playSec(playSec)
                                .positionSec(positionSec)
                                .videoId(videoId)
                                .ts(ts)
                                .build();
                    }
                }
        );

        //maplp.print();
        SingleOutputStreamOperator<TableLearningPlay> rbq = maplp.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TableLearningPlay>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TableLearningPlay>() {
                                    @Override
                                    public long extractTimestamp(TableLearningPlay tbp, long l) {
                                        return tbp.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<TableLearningPlay, String> tablekeyby = rbq.keyBy(
                new KeySelector<TableLearningPlay, String>() {
                    @Override
                    public String getKey(TableLearningPlay tb) throws Exception {
                        return tb.getSessionId();
                    }
                }
        );

        WindowedStream<TableLearningPlay, String, TimeWindow> window = tablekeyby.window(EventTimeSessionWindows.withGap(Time.seconds(3L)));

        SingleOutputStreamOperator<TableLearningPlay> reduce = window.reduce(
                new ReduceFunction<TableLearningPlay>() {
                    @Override
                    public TableLearningPlay reduce(TableLearningPlay t1, TableLearningPlay t2) throws Exception {
                        t1.setPlaySec(t1.getPlaySec() + t2.getPlaySec());
                        if (t2.getTs() > t1.getTs()) {
                            t1.setPositionSec(t2.getPositionSec());
                        }
                        return t1;
                    }
                },
                new ProcessWindowFunction<TableLearningPlay, TableLearningPlay, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TableLearningPlay, TableLearningPlay, String, TimeWindow>.Context context, Iterable<TableLearningPlay> iterable, Collector<TableLearningPlay> collector) throws Exception {

                        for (TableLearningPlay tableLearningPlay : iterable) {
                            collector.collect(tableLearningPlay);
                        }

                    }
                }
        );
        //reduce.print();


        SingleOutputStreamOperator<String> map = reduce.map(
                new MapFunction<TableLearningPlay, String>() {
                    @Override
                    public String map(TableLearningPlay tableLearningPlay) throws Exception {
                        String jsonString = JSON.toJSONString(tableLearningPlay);
                        return jsonString;
                    }
                }
        );
        map.print();

        map.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_LEARNING_PLAY));
    }
}
